package com.bestv.bsr.cloud.service.impl.ah;

import cn.hutool.core.collection.CollUtil;
import com.bestv.bsr.cloud.common.BsrThreadFactory;
import com.bestv.bsr.cloud.common.DisplayNotifyService;
import com.bestv.bsr.core.annotations.OptionalComponent;
import com.bestv.bsr.core.constants.AiSourceConstants;
import com.bestv.bsr.core.constants.CacheConstants;
import com.bestv.bsr.core.constants.ServiceCodeEnum;
import com.bestv.bsr.core.context.RecommendContext;
import com.bestv.bsr.core.entities.vo.RecDataVo;
import com.bestv.bsr.core.exceptions.ServiceException;
import com.bestv.bsr.core.kafka.BSRMsg;
import com.bestv.bsr.core.kafka.MsgType;
import com.bestv.bsr.core.kafka.dto.RecommendRefreshBody;
import com.bestv.bsr.core.service.impl.ah.AbstractAHRecommendEngine;
import com.bestv.bsr.core.service.impl.ah.dto.AHRequestParam;
import com.bestv.bsr.core.service.impl.ah.dto.AHResponseDto;
import com.bestv.bsr.core.service.impl.ah.dto.AHZYWLResponseDto;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.skywalking.apm.toolkit.trace.RunnableWrapper;
import org.redisson.api.RBucket;
import org.redisson.api.RedissonClient;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * cloud ah recommend engine
 *
 * @author taojiacheng
 */
@Slf4j
@RequiredArgsConstructor
@OptionalComponent(value = "ahRecommendEngine", strategy = OptionalComponent.Strategy.CLOUD)
public class CloudAHRecommendEngine extends AbstractAHRecommendEngine {

    protected final DisplayNotifyService displayNotifyService;
    protected final RedissonClient redissonClient;
    protected final BsrThreadFactory bsrThreadFactory;

    @Override
    public RecDataVo recommend(RecommendContext recommendContext) {
        AHRequestParam ahRequestParam = buildRequestParam(recommendContext, CollUtil.isNotEmpty(recommendContext.getItemCodes()));
        AHZYWLResponseDto responseDto;
        AHResponseDto ahResponseDto;
        // 缓存开启
        if (properties.isCacheEnabled()) {
            String cacheKey = formatCacheKey(recommendContext);
            RBucket<AHZYWLResponseDto> bucket = redissonClient.getBucket(cacheKey);
            responseDto = bucket.get();
            if (responseDto == null) {
                responseDto = ahRemoteService.recommend(ahRequestParam,
                        (result, serverContext, methodContext, paramContext) -> {
                            // 调用后 [降级之前] 回调
                            cacheAndNotifyDisplayService(cacheKey, result, recommendContext);
                        }
                );
            }
        } else {
            // 缓存未开启
            responseDto = ahRemoteService.recommend(ahRequestParam);
            cacheAndNotifyDisplayService("", responseDto, recommendContext);
        }
        ahResponseDto = handlerZYWLResp(responseDto);
        return convertResult(ahResponseDto, recommendContext);
    }


    /**
     * 生成缓存并通知展现服务
     */
    protected void cacheAndNotifyDisplayService(String cacheKey, Object result, RecommendContext recommendContext) {
        bsrThreadFactory.execute(RunnableWrapper.of(() -> {
                    if (result instanceof AHZYWLResponseDto) {
                        AHZYWLResponseDto resp = (AHZYWLResponseDto) result;
                        if (resp.getCode() != 200) {
                            return;
                        }
                        // 通知
                        try {
                            // 通知逻辑 copy from 长征 bsr-service
                            List<String> itemCodes = resp.getData().getCates().get(0).getItems().stream()
                                    .filter(itemDto -> properties.getCpid().equals(itemDto.getCpid()))
                                    .map(AHZYWLResponseDto.ZYWLRespItems::getItemid)
                                    .collect(Collectors.toList());
                            BSRMsg.Header header = new BSRMsg.Header(MsgType.RECOMMEND_REFRESH, MsgType.RECOMMEND_REFRESH.getName() + ":" + cacheKey);
                            RecommendRefreshBody body = RecommendRefreshBody.builder()
                                    .userGroupCode(recommendContext.getUserGroupCode())
                                    .ruleCode(recommendContext.getRuleCode())
                                    .itemCodes(itemCodes)
                                    .aiSource(recommendContext.getAiSource())
                                    .build();
                            BSRMsg<RecommendRefreshBody> bsrMsg = new BSRMsg<>(header, body);
                            displayNotifyService.notify(bsrMsg);
                        } catch (Exception e) {
                            log.error("notify error: {}", e.getMessage(), e);
                        }
                        if (properties.isCacheEnabled() && StringUtils.isNotBlank(cacheKey)) {
                            // 缓存
                            RBucket<AHZYWLResponseDto> bucket = redissonClient.getBucket(cacheKey);
                            bucket.set(resp, properties.getExpireSecond(), TimeUnit.SECONDS);
                        }
                    }
                })
        );
    }

    /**
     * 生成缓存key
     */
    protected String formatCacheKey(RecommendContext recommendContext) {
        return CacheConstants.format(CacheConstants.AH_CACHE_TEMPLATE,
                recommendContext.getAiSource(),
                recommendContext.getChannel(),
                recommendContext.getCateName(),
                // todo 取0？
                CollUtil.isNotEmpty(recommendContext.getItemCodes()) ? recommendContext.getItemCodes().get(0) : "",
                recommendContext.getUserAccount(),
                recommendContext.getDeviceid()
        );
    }


    /**
     * todo 看看能不能落到独立配置文件中
     * 安徽驻地映射关系梳理
     * 首页推荐场景 -> sceneId:7 -> channel:1 & ahCodes[前端传入的ahCode]|cateName:99[others]
     * 点播详情页场景 -> sceneId:1 -> channel:2 & cateName:98[others]
     * 收藏历史推荐位 -> sceneId:history_norecord_v7|sceneId:10 -> channel:3 & cateName:98[all]
     * 退出挽留推荐位 -> sceneId:2|sceneId:3 -> channel:2 & cateName:98[all]
     */
    @Override
    public void init(RecommendContext recommendContext) {
        super.init(recommendContext);
        if ("7".equals(recommendContext.getSceneId())) {
            List<String> ahCodes = recommendContext.getAhCodes();
            if (CollUtil.isNotEmpty(ahCodes)) {
                recommendContext.setCateName(ahCodes.get(0));
            } else {
                recommendContext.setCateName(properties.getItemTypeMap().get("others"));
            }
            return;
        }
        recommendContext.setCateName(properties.getItemTypeMap().get("all"));
    }

    @Override
    public void validate(RecommendContext recommendContext) {
        // 参数校验
        String mac = recommendContext.getMac();
        if (StringUtils.isBlank(mac)) {
            throw new ServiceException(ServiceCodeEnum.MAC_IS_BLANK);
        }
        String partnerUserAccount = recommendContext.getPartnerUserAccount();
        if (StringUtils.isBlank(partnerUserAccount)) {
            throw new ServiceException(ServiceCodeEnum.USER_ACCOUNT_IS_BLANK);
        }
    }

    @Override
    protected List<String> addSupportedAiSource() {
        return Collections.singletonList(AiSourceConstants.AI_SOURCE_ANYD_ZN);
    }
}
